package demoflink.process;

import demoflink.entity.WaterSensor;
import demoflink.function.WaterSensorMapFunction;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class KeyedProcessTimerDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> sensorDs = env.socketTextStream("node1", 7777)
                .map(new WaterSensorMapFunction());
        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofMillis(3))
                .withTimestampAssigner(((element, recordTimestamp) -> {
                    System.out.println("数据=" + element + ",recordTimestamp=" + recordTimestamp);
                    return element.getTs() * 1000L;
                }));
        SingleOutputStreamOperator<WaterSensor> sensorSingleOutputStreamOperator = sensorDs.assignTimestampsAndWatermarks(watermarkStrategy);
        KeyedStream<WaterSensor, String> keyBy = sensorSingleOutputStreamOperator.keyBy(sensor -> sensor.getId());
        keyBy.process(new KeyedProcessFunction<String, WaterSensor, String>() {
            @Override
            public void processElement(WaterSensor waterSensor, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> collector) throws Exception {
                String currentKey = ctx.getCurrentKey();
                //数据中提取出来的事件时间
                Long timestamp = ctx.timestamp();
                System.out.println("当前时间是"+timestamp+",注册了一个5s的定时器");
                TimerService timerService = ctx.timerService();
                //事件时间
                timerService.registerEventTimeTimer(5000L);
                // timerService.registerProcessingTimeTimer(timestamp + 1L);
                // timerService.deleteEventTimeTimer(timestamp);
                // timerService.deleteProcessingTimeTimer(timestamp + 1);

                timerService.currentProcessingTime();
                timerService.currentWatermark();
            }

            /**
             * 时间进展到定时器注册的时间，调同该方法
             * @param timestamp
             * @param ctx
             * @param out
             * @throws Exception
             */
            @Override
            public void onTimer(long timestamp, KeyedProcessFunction<String, WaterSensor, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                super.onTimer(timestamp, ctx, out);
                System.out.println("现在时间是"+timestamp+"定时器触发");
            }
        }).print();
        env.execute();

    }
}
